package io.renren.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.renren.config.TaskExecutorConfig;
import io.renren.modules.wmstate.dao.WeldingmachineStateDao;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
//import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.EnableRetry;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * 焊机实时数据接口
 *
 * */

@Slf4j
@Configuration
@EnableRetry // 启用 Spring 的重试功能
public class MqttConfiguration {
    //定义一个暂时存储获取到的pointData的集合
    private static List<PointData> pointDataList = new ArrayList<>();
    @Autowired
    MyWebSocketHandler myWebSocketHandler;

    @Autowired

    private ThreadPoolTaskExecutor taskExecutor;

    @Autowired
    WeldingmachineStateDao weldingmachineStateDao;

    static String clientId = UUID.randomUUID().toString().replace("-", "");
    private static final String URL = "tcp://10.142.3.129:1883";
//    private static final String URL = "tcp://weld.cmhijs.cn:1883";
    private static final String CLIENT_ID = clientId;
    private static final String TOPIC = "/gw/acrelHW/AWT100/data/#";

    private static final String TOPIC2 = "/delivery/welstatus/#";


    private static final int MAX_RETRY_ATTEMPTS = 5;
    private int retryCount = 0;

    @Bean
    public MqttConnectOptions mqttConnectOptions(){
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setServerURIs(new String[]{URL});
        mqttConnectOptions.setConnectionTimeout(100);
        mqttConnectOptions.setAutomaticReconnect(true); // 设置自动重连
        mqttConnectOptions.setCleanSession(true);
        return mqttConnectOptions;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel mqttInputChannel2() {
        return new DirectChannel();
    }

    @Bean
    @Retryable(value = {MqttException.class}, maxAttempts = MAX_RETRY_ATTEMPTS, backoff = @Backoff(delay = 5000))
    public MessageProducer inbound() {
        CustomMqttPahoMessageDrivenChannelAdapter adapter =
                new CustomMqttPahoMessageDrivenChannelAdapter(CLIENT_ID, mqttClientFactory(), TOPIC);
        adapter.setCompletionTimeout(50000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Value("${mqtt.enable}")
    private boolean mqttEnabled; // 通过配置文件注入是否启用 MQTT 订阅的值

    @Bean
    @Retryable(value = {MqttException.class}, maxAttempts = MAX_RETRY_ATTEMPTS, backoff = @Backoff(delay = 5000))
    public MessageProducer inbound2() {
        if (mqttEnabled) {
            CustomMqttPahoMessageDrivenChannelAdapter adapter =
                    new CustomMqttPahoMessageDrivenChannelAdapter(CLIENT_ID + "_2", mqttClientFactory(), TOPIC2);
            adapter.setCompletionTimeout(50000);
            adapter.setConverter(new DefaultPahoMessageConverter());
            adapter.setQos(0);
            adapter.setOutputChannel(mqttInputChannel2());
            return adapter;
        } else {
            return null;
        }
    }

    @Recover
    public void recover(MqttException ex) {
        if (retryCount < MAX_RETRY_ATTEMPTS) {
            log.error("连接到 MQTT 代理失败。正在进行第 " + (retryCount + 1) + " 次重试...", ex);
            retryCount++;
        } else {
            log.error("连接到 MQTT 代理失败，重试次数已达到上限。执行定时任务进行重试或通知相关人员。", ex);
            // 这里您可以执行定时任务进行重试，或者触发通知机制
            // 例如：scheduleRetry();
        }
    }

    @Scheduled(fixedDelay = 60000) // 每隔一分钟执行一次定时任务
    public void scheduleRetry() {
        if (retryCount >= MAX_RETRY_ATTEMPTS) {
            // 在定时任务中进行重试或者执行通知机制
            log.info("执行定时任务进行 MQTT 连接重试...");
            // 重置重试计数器
            retryCount = 0;
            // 进行新的连接尝试，您可以调用相关方法重新尝试连接
            // retryConnection();
        }
    }

    /**
     * 定位方式
     * 经度
     * 纬度
     * time (时间戳格式)
     * 信号强度
     * 总功率因数
     * 正向有功总电能示值
     * 设备编号
     * 通电状态
     * AB线电压
     * 设备状态
     * A相电流
     * 总有功功率
     * */
@Data
public static class PointDataNew{
@TableId(type=IdType.ASSIGN_UUID)
        private  String id;
        private  String gps_type;
        private  String time;
        private  double latitude;
        private  int signal_strength;
        private double longitude;
        private  double total_power_factor;
        private  double forward_active_energy_indication;
        private  double device_id;
        private  String power_state;
        private  double phase_voltage;
        private  String device_state;
        private  double phase_current;
        private  double total_power;

    }




    @Data
    public static class PointData {
        @TableId(type = IdType.ASSIGN_UUID)
        private String id;
        private String type;
        private String time;
        private String datatime;
        private String gwSN;
        private String meterSN;
        private String meterName;
        private String meterStatus;
        private int ch;
        private double U;
        private double I;
        private double P;
        private int TEMP;
        private double EPI;
        private double EPE;
        private int PT;
        private int CT;
    }


    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) throws Exception {
        String msg = (String) message.getPayload();
        //定义一个常量120，用于修改电流I的值
        final int a = 120;
        //json字符串返回值反序列化为实体类
        PointData pointData = JSONObject.parseObject(msg,PointData.class);

        // 检查电压U的值，如果不为0，则修改电流I的值
        if (pointData.getU() != 0) {
            pointData.setI(pointData.getI() * a);
        }

        //设置uuid作为主键
        pointData.setId(UUID.randomUUID().toString());


        myWebSocketHandler.sendMessage((String) message.getPayload());

        //将pointData存入集合
        pointDataList.add(pointData);
        //判断pointDataList的长度是否为500，如果是则将pointDataList批量存入数据库
        if(pointDataList.size()==500){
            List<PointData> tempList = new ArrayList<>(pointDataList);
            pointDataList.clear();
//            log.info("批量插入线程-" + Thread.currentThread().getId() + "正在执行");
//            taskExecutor.execute(() -> weldingmachineStateDao.insertBatch(tempList));
//            weldingmachineStateDao.insertBatch(tempList);
        }
    }

    private int workingCount = 0; // 工作中的设备数量
    private int notWorkingCount = 0; // 故障中的设备数量
    private final int totalCount = 100; // 离线中的设备数量

    @ServiceActivator(inputChannel = "mqttInputChannel2")
    public void handleMessage2(Message<?> message) throws Exception {
        String msg = (String) message.getPayload();
        // 处理第二个主题的消息，消息示例：{"wmNo":"72","gatew":"12304178610212","status":"0"}
        myWebSocketHandler.sendMessage((String) message.getPayload());
    }
}